#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__
/*异步工作器的实现
 */

#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <atomic>
namespace PRJLOG
{
    enum class Asyncloopertype
    {
        ASYNC_SAFE,  // 安全表示容量固定，不会将空间耗尽
        ASYNC_UNSAFE // 不安全表示容量扩容
    };
    using functor = std::function<void(Buffer &)>;
    class Asynclooper
    {
    public:
        using ptr = std::shared_ptr<Asynclooper>;
        Asynclooper(const functor &cb, Asyncloopertype loopertype = Asyncloopertype::ASYNC_SAFE)
            : _callback(cb), _stop(false), _thread(std::thread(&Asynclooper::threadwork, this)), _Asynctype(loopertype)
        {
        }
        ~Asynclooper() { stop(); }
        // 异步工作器停止
        void stop()
        {
            _stop = true;
            _cond_con.notify_all(); // 唤醒消费者线程
            _thread.join();         // 回收线程
        }
        void push(const char *data, size_t len)
        {
            // 1.固定容量2.动态扩容
            // 判断_pro_buf是否可以push
            std::unique_lock<std::mutex> lock(_mutex);
            if (_Asynctype == Asyncloopertype::ASYNC_SAFE)
            {
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writerablesize() >= len; });
            }
            _pro_buf.push(data, len);
            // 唤醒消费者线程
            _cond_con.notify_one();
        }

    private:
        // 异步工作器的线程工作入口,是消费者线程对数据的处理
        void threadwork()
        {
            while (1)
            {
                {
                    // 1.判断生产者缓冲区是否有数据，有则交换，无则阻塞
                    std::unique_lock<std::mutex> lock(_mutex);
                    if (_stop && _pro_buf.empty())
                        break;
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });
                    _con_buf.swapbuffer(_pro_buf);
                    // 2.唤醒生产者线程
                    if (_Asynctype == Asyncloopertype::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 3.唤醒后处理数据
                _callback(_con_buf);
                // 4.初始化缓冲区
                _con_buf.reset();
            }
        }

    private:
        functor _callback; // 回调函数，为消费线程对数据的处理，由异步工作器使用者传入
    private:
        Asyncloopertype _Asynctype;        // 容量安全
        std::atomic<bool> _stop;           // 异步工作器的工作标志
        Buffer _pro_buf;                   // 生产者缓冲区
        Buffer _con_buf;                   // 消费者缓冲区
        std::mutex _mutex;                 // 互斥锁
        std::condition_variable _cond_pro; // 生产者条件变量
        std::condition_variable _cond_con; // 消费者条件变量
        std::thread _thread;               // 异步工作器的工作线程
    };
}

#endif